package chapter3.eg4;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.apache.avro.Schema;
import org.apache.avro.Schema.Parser;
import org.apache.avro.generic.GenericData.Record;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

public class GenericAvroApp {

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    Properties properties = new Properties();
    properties.put("bootstrap.servers", "localhost:9092");
    properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    properties.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
    properties.put("schema.registry.url", "http://localhost:8081");
    String schemaString = "{\n"
        + "  \"namespace\": \"customerManagement.avro\",\n"
        + "  \"type\": \"record\",\n"
        + "  \"name\": \"Customer\",\n"
        + "  \"fields\": [\n"
        + "    {\"name\": \"id\", \"type\": \"int\"},\n"
        + "    {\"name\": \"name\", \"type\": \"string\"},\n"
        + "    {\"name\": \"email\", \"type\": \"string\"}\n"
        + "  ]\n"
        + "}";

    Producer<String, GenericRecord> producer = new KafkaProducer<>(properties);
    Schema.Parser parser = new Parser();
    Schema schema = parser.parse(schemaString);

    while (true) {
      GenericRecord customer = new Record(schema);
      customer.put("id", 1);
      customer.put("name", "zhangsan");
      customer.put("email", "909012366@qq.com");
      ProducerRecord<String, GenericRecord> data = new ProducerRecord<>("customerContacts", "exampleCustomer", customer);
      Future<RecordMetadata> future = producer.send(data);
      System.out.println(future.get().partition());
      TimeUnit.SECONDS.sleep(1);
    }
  }

}
